package com.zvz.RxJava;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.util.Arrays;

public class Demo1 {

    public static void main(String[] args) {
        // hello("1", "2", "3");

        // 创建一个Observable  事件生产
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            public void call(Subscriber<? super String> subscriber) {        // 发送一个 Hello World 事件
                subscriber.onNext("Hello World!");        // 事件发送完成
                subscriber.onCompleted();
            }
        });

        Observable<String> observable1 = Observable.just("Hello World!");

        // 创建一个Subscriber
        Subscriber<String> subscriber = new Subscriber<String>() {
            public void onCompleted() {
                System.out.println("complete");
            }

            public void onError(Throwable e) {
                System.out.println("onError");
            }

            public void onNext(String s) {
                System.out.println(s);
            }
        };
        // 订阅事件
        observable.subscribe(subscriber);

        // onComplete()
        Action0 onCompleteAction = new Action0() {
            public void call() {
                System.out.println("onCompleteAction");
            }
        };
        // onNext(T t)
        Action1<String> onNextAction = new Action1<String>() {
            public void call(String s) {
                System.out.println("onNextAction");
                System.out.println(s);
            }
        };
        //onError(Throwable t)
        Action1<Throwable> onErrorAction = new Action1<Throwable>() {
            public void call(Throwable throwable) {
                System.out.println("onErrorAction");
            }
        };

        //RxJava 的事件订阅支持以下三种不完整定义的回调。
        /**
         * 我们可以根据当前需要，传入对应的 Action， RxJava 会相应的自动创建 Subscriber。
         * Action0 表示一个无回调参数的Action；
         * Action1 表示一个含有一个回调参数的Action；
         * 当然，还有Action2 ~ Action9，分别对应2~9个参数的Action；
         * 每个Action，都有一个 call() 方法，通过泛型T，来指定对应参数的类型；
         */
        observable.subscribe(onNextAction);

        observable.subscribe(onNextAction, onErrorAction);

        observable.subscribe(onNextAction, onErrorAction, onCompleteAction);


        threadChange();

    }

    /**
     * 在 RxJava 中，提供了一个名为 Scheduler 的线程调度器，RxJava 内部提供了4个调度器，分别是：
     * Schedulers.io(): I/O 操作（读写文件、数据库、网络请求等），与newThread()差不多，区别在于io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 效率比 newThread() 更高。值得注意的是，在 io() 下，不要进行大量的计算，以免产生不必要的线程；
     * Schedulers.newThread(): 开启新线程操作；
     * Schedulers.immediate(): 默认指定的线程，也就是当前线程；
     * Schedulers.computation():计算所使用的调度器。这个计算指的是 CPU 密集型计算，即不会被 I/O等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。值得注意的是，不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU；
     * AndroidSchedulers.mainThread(): RxJava 扩展的 Android 主线程；
     *
     * 我们可以通过 subscribeOn() 和 observeOn() 这两个方法来进行线程调度。
     */
    public static void threadChange(){
        Observable
                .just(0)
                .observeOn(Schedulers.computation())
                .map(i -> i + 1)
                .subscribeOn(Schedulers.newThread())
                .subscribe(System.out::println);
    }

    /**
     *map 与 flatMap 这两个操作符的共同点在于，他们都是把一个对象转换为另一个对象，但须注意以下这些特点:
     * flatMap 返回的是一个Observable对象，而 map 返回的是一个普通转换后的对象;
     * flatMap 返回的Observable对象并不是直接发送到Subscriber的回调中，而是重新创建一个Observable对象，并激活这个Observable对象，使之开始发送事件；而 map 变换后返回的对象直接发到Subscriber回调中；
     * flatMap 变换后产生的每一个Observable对象发送的事件，最后都汇入同一个Observable，进而发送给Subscriber回调；
     * map返回类型 与 flatMap 返回的Observable事件类型，可以与原来的事件类型一样；
     * 可以对一个Observable多次使用 map 和 flatMap；
     * 鉴于 flatMap 自身强大的功能，这常常被用于 嵌套的异步操作，例如嵌套网络请求。传统的嵌套请求，一般都是在前一个请求的 onSuccess() 回调里面发起新的请求，这样一旦嵌套多个的话，缩进就是大问题了，而且严重的影响代码的可读性。而RxJava嵌套网络请求仍然通过链式结构，保持代码逻辑的清晰
     */


    public static void hello(String... names) {
        Observable.from(names).subscribe(new Action1<String>() {
            public void call(String s) {
                System.out.println("Hello " + s + "!");
            }

        });

        Observable.create(new Observable.OnSubscribe<String>(){
            public void call(Subscriber<? super String> subscriber) {

            }
        }).map(new Func1<String, Integer>() {
            public Integer call(String s) {
                return null;
            }
        }).subscribe();
    }
}
